extern crate core;

use crate::rtmp::relay::pull_client::PullClient;
use crate::rtmp::relay::push_client::PushClient;
use crate::rtmp::RtmpServer;
use crate::rtsp::RtspServer;
use log::LevelFilter;
use std::io::Write;
use streamhub::StreamsHub;

mod rtmp;
mod rtsp;

fn start_single_server() {
    let mut stream_hub = StreamsHub::new(None);
    let sender = stream_hub.get_hub_event_sender();

    let listen_port = 1935;
    let address = format!("0.0.0.0:{port}", port = listen_port);

    let mut rtmp_server = RtmpServer::new(address, sender, 1usize, None);
    tokio::spawn(async move {
        if let Err(err) = rtmp_server.start().await {
            log::error!("rtmp server error: {}\n", err);
        }
    });

    tokio::spawn(async move { stream_hub.run().await });
}

fn start_cluster_server(listen_port: u16, push_nodes: Vec<String>, pull_nodes: Vec<String>) {
    let mut stream_hub = StreamsHub::new(None);
    let sender = stream_hub.get_hub_event_sender();

    // 配置推流客户端（到其他集群节点）
    for node in push_nodes {
        let mut push_client = PushClient::new(
            node.to_string(),
            stream_hub.get_client_event_consumer(),
            sender.clone(),
        );

        tokio::spawn(async move {
            if let Err(err) = push_client.run().await {
                log::error!("Push to {} failed: {}", node, err);
            }
        });
    }
    stream_hub.set_rtmp_push_enabled(true);

    // 配置拉流客户端（从其他集群节点）
    for node in pull_nodes {
        let mut pull_client = PullClient::new(
            node.to_string(),
            stream_hub.get_client_event_consumer(),
            sender.clone(),
        );

        tokio::spawn(async move {
            if let Err(err) = pull_client.run().await {
                log::error!("Pull from {} failed: {}", node, err);
            }
        });
    }
    stream_hub.set_rtmp_pull_enabled(true);

    // 启动本节点RTMP服务
    let listen_address = format!("0.0.0.0:{port}", port = listen_port);
    let mut rtmp_server = RtmpServer::new(listen_address, sender.clone(), 1, None);

    tokio::spawn(async move {
        if let Err(err) = rtmp_server.start().await {
            log::error!("RTMP server failed: {}", err);
        }
    });

    tokio::spawn(async move { stream_hub.run().await });
}

fn start_rtsp() {
    let mut stream_hub = StreamsHub::new(None);
    let sender = stream_hub.get_hub_event_sender();

    // 获取 RTSP 服务器监听的端口号
    let listen_port = 554;
    // 构造 RTSP 服务器监听的地址，使用 0.0.0.0 表示监听所有可用的网络接口
    let address = format!("0.0.0.0:{port}", port = listen_port);

    // 创建一个新的 RTSP 服务器实例
    let mut rtsp_server = RtspServer::new(address, sender, None);

    // 使用 tokio::spawn 异步启动 RTSP 服务器
    tokio::spawn(async move {
        if let Err(err) = rtsp_server.start().await {
            log::error!("rtsp server error: {}\n", err);
        }
    });
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    env_logger::Builder::from_default_env()
        .filter_level(log::LevelFilter::Debug)
        .init();

    // 启动 rtsp 服务器
    start_rtsp();

    // log::info!("Starting cluster Server...");
    //
    // // 获取用户输入
    // let listening_port = get_user_input("请输入要监听的端口: ");
    // let push_targets = get_user_input("请输入推送目标的 IP 和端口（用逗号分隔多个目标）: ");
    // let pull_sources = get_user_input("请输入拉取目标的 IP 和端口（用逗号分隔多个目标）: ");
    //
    // // 解析用户输入
    // let listening_port: u16 = listening_port.parse()?;
    // let push_targets: Vec<String> = push_targets
    //     .split(',')
    //     .map(|s| s.trim().to_string())
    //     .collect();
    // let pull_sources: Vec<String> = pull_sources
    //     .split(',')
    //     .map(|s| s.trim().to_string())
    //     .collect();
    //
    // // 启动集群服务器
    // start_cluster_server(listening_port, push_targets, pull_sources);

    // 等待 Ctrl+C 信号
    tokio::signal::ctrl_c().await?;
    Ok(())
}

fn get_user_input(prompt: &str) -> String {
    print!("{}", prompt);
    std::io::stdout().flush().unwrap(); // 确保提示显示出来

    let mut input = String::new();
    std::io::stdin().read_line(&mut input).unwrap();
    input.trim().to_string()
}
